/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Meiyi & Wangyunlai on 2021/5/13.
//

#include <limits.h>
#include <locale>
#include <string.h>
#include <filesystem>

#include "common/defs.h"
#include "common/lang/string.h"
#include "common/lang/span.h"
#include "common/lang/algorithm.h"
#include "common/log/log.h"
#include "common/global_context.h"
#include "common/rc.h"
#include "sql/operator/physical_operator.h"
#include "sql/parser/parse_defs.h"
#include "storage/db/db.h"
#include "storage/buffer/disk_buffer_pool.h"
#include "storage/common/condition_filter.h"
#include "storage/common/meta_util.h"
#include "storage/field/field_meta.h"
#include "storage/index/bplus_tree.h"
#include "storage/index/bplus_tree_index.h"
#include "storage/index/index.h"
#include "storage/index/index_meta.h"
#include "storage/record/record_manager.h"
#include "storage/table/table.h"
#include "storage/trx/trx.h"
#include "sql/operator/project_physical_operator.h"

Table::~Table()
{
  if (record_handler_ != nullptr) {
    delete record_handler_;
    record_handler_ = nullptr;
  }

  if (data_buffer_pool_ != nullptr) {
    data_buffer_pool_->close_file();
    data_buffer_pool_ = nullptr;
  }

  for (vector<Index *>::iterator it = indexes_.begin(); it != indexes_.end(); ++it) {
    Index *index = *it;
    delete index;
  }
  indexes_.clear();

  LOG_INFO("Table has been closed: %s", name());
}

RC Table::create(Db *db, int32_t table_id, const char *path, const char *name, const char *base_dir,
    span<const AttrInfoSqlNode> attributes, StorageFormat storage_format)
{
  if (table_id < 0) {
    LOG_WARN("invalid table id. table_id=%d, table_name=%s", table_id, name);
    return RC::INVALID_ARGUMENT;
  }

  if (common::is_blank(name)) {
    LOG_WARN("Name cannot be empty");
    return RC::INVALID_ARGUMENT;
  }
  LOG_INFO("Begin to create table %s:%s", base_dir, name);

  if (attributes.size() == 0) {
    LOG_WARN("Invalid arguments. table_name=%s, attribute_count=%d", name, attributes.size());
    return RC::INVALID_ARGUMENT;
  }

  RC rc = RC::SUCCESS;

  // 使用 table_name.table记录一个表的元数据
  // 判断表文件是否已经存在
  // 下面是创建了一个.table文件
  int fd = ::open(path, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
  if (fd < 0) {
    if (EEXIST == errno) {
      LOG_ERROR("Failed to create table file, it has been created. %s, EEXIST, %s", path, strerror(errno));
      return RC::SCHEMA_TABLE_EXIST;
    }
    LOG_ERROR("Create table file failed. filename=%s, errmsg=%d:%s", path, errno, strerror(errno));
    return RC::IOERR_OPEN;
  }

  close(fd);

  // 创建文件
  const vector<FieldMeta> *trx_fields = db->trx_kit().trx_fields();
  if ((rc = table_meta_.init(table_id, name, trx_fields, attributes, storage_format)) != RC::SUCCESS) {
    LOG_ERROR("Failed to init table meta. name:%s, ret:%d", name, rc);
    return rc;  // delete table file
  }

  fstream fs;
  fs.open(path, ios_base::out | ios_base::binary);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open file for write. file name=%s, errmsg=%s", path, strerror(errno));
    return RC::IOERR_OPEN;
  }

  // 记录元数据到.table文件中
  table_meta_.serialize(fs);
  fs.close();

  db_       = db;
  base_dir_ = base_dir;

  string             data_file = table_data_file(base_dir, name);  // 取得.data文件的绝对路径
  BufferPoolManager &bpm       = db->buffer_pool_manager();
  rc                           = bpm.create_file(data_file.c_str());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create disk buffer pool of data file. file name=%s", data_file.c_str());
    return rc;
  }

  rc = init_record_handler(base_dir);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create table %s due to init record handler failed.", data_file.c_str());
    // don't need to remove the data_file
    return rc;
  }

  LOG_INFO("Successfully create table %s:%s", base_dir, name);
  return rc;
}

// 重载一个建表的函数，用来支持create table as select ...
RC Table::create(Db *db, int32_t table_id, const char *path, const char *name, const char *base_dir,
    unique_ptr<Expression> &select_expression, span<const AttrInfoSqlNode> attributes, StorageFormat storage_format)
{
  if (table_id < 0) {
    LOG_WARN("invalid table id. table_id=%d, table_name=%s", table_id, name);
    return RC::INVALID_ARGUMENT;
  }

  if (common::is_blank(name)) {
    LOG_WARN("Name cannot be empty");
    return RC::INVALID_ARGUMENT;
  }
  LOG_INFO("Begin to create table %s:%s", base_dir, name);

  vector<vector<Value>> rows;
  SubQueryExpr         *select_expr       = static_cast<SubQueryExpr *>(select_expression.get());
  PhysicalOperator     *physical_operator = select_expr->get_physical_operator().get();
  RC                    rc                = physical_operator->open(nullptr);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to open physical operator. ret=%d", rc);
    return rc;
  }
  while (RC::SUCCESS == (rc = physical_operator->next())) {
    Tuple *tuple = physical_operator->current_tuple();
    if (nullptr == tuple) {
      LOG_ERROR("Failed to get current tuple. ret=%d", rc);
      physical_operator->close();
      return RC::INTERNAL;
    }
    vector<Value> row;
    for (int i = 0; i < tuple->cell_num(); i++) {
      Value value;
      rc = tuple->cell_at(i, value);
      if (rc != RC::SUCCESS) {
        LOG_ERROR("Failed to get cell value. ret=%d", rc);
        physical_operator->close();
        return rc;
      }
      row.push_back(value);
    }
    rows.push_back(row);
  }
  if (rc != RC::RECORD_EOF) {
    LOG_ERROR("Failed to get next tuple. ret=%d", rc);
    physical_operator->close();
    return rc;
  }
  rc = physical_operator->close();
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to close physical operator. ret=%d", rc);
    return rc;
  }

  // 这个用来存放select中的属性信息
  vector<AttrInfoSqlNode> new_attributes;

  // 这个变量用来判断
  span<const AttrInfoSqlNode> need_attributes;

  ProjectPhysicalOperator *project_physical_operator = static_cast<ProjectPhysicalOperator *>(physical_operator);
  for (unsigned i = 0; i < project_physical_operator->expressions().size(); i++) {
    AttrInfoSqlNode attr;
    attr.type     = project_physical_operator->expressions()[i]->value_type();
    attr.length   = project_physical_operator->expressions()[i]->value_length();
    attr.nullable = true;
    // 名字应该从select_expr中获取
    attr.name = project_physical_operator->expressions()[i]->name();
    new_attributes.push_back(attr);
  }

  if (new_attributes.size() == 0) {
    LOG_WARN("Invalid arguments. table_name=%s, attribute_count=%d", name, attributes.size());
    return RC::INVALID_ARGUMENT;
  }
  // 如果用户指定了属性，应该使用用户指定的属性来创建表,否则的话使用查询出来的结果的属性
  if (attributes.size() != 0) {
    need_attributes = attributes;
  } else {
    need_attributes = span<const AttrInfoSqlNode>(new_attributes);
  }

  rc = RC::SUCCESS;

  // 使用 table_name.table记录一个表的元数据
  // 判断表文件是否已经存在
  // 下面是创建了一个.table文件
  int fd = ::open(path, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
  if (fd < 0) {
    if (EEXIST == errno) {
      LOG_ERROR("Failed to create table file, it has been created. %s, EEXIST, %s", path, strerror(errno));
      return RC::SCHEMA_TABLE_EXIST;
    }
    LOG_ERROR("Create table file failed. filename=%s, errmsg=%d:%s", path, errno, strerror(errno));
    return RC::IOERR_OPEN;
  }

  close(fd);

  // 创建文件
  const vector<FieldMeta> *trx_fields = db->trx_kit().trx_fields();
  if ((rc = table_meta_.init(table_id, name, trx_fields, need_attributes, storage_format)) != RC::SUCCESS) {
    LOG_ERROR("Failed to init table meta. name:%s, ret:%d", name, rc);
    return rc;  // delete table file
  }

  fstream fs;
  fs.open(path, ios_base::out | ios_base::binary);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open file for write. file name=%s, errmsg=%s", path, strerror(errno));
    return RC::IOERR_OPEN;
  }

  // 记录元数据到.table文件中
  table_meta_.serialize(fs);
  fs.close();

  db_       = db;
  base_dir_ = base_dir;

  string             data_file = table_data_file(base_dir, name);  // 取得.data文件的绝对路径
  BufferPoolManager &bpm       = db->buffer_pool_manager();
  rc                           = bpm.create_file(data_file.c_str());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create disk buffer pool of data file. file name=%s", data_file.c_str());
    return rc;
  }

  rc = init_record_handler(base_dir);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create table %s due to init record handler failed.", data_file.c_str());
    // don't need to remove the data_file
    return rc;
  }

  LOG_INFO("Successfully create table %s:%s", base_dir, name);

  // 创建成功之后，开始插入数据
  for (unsigned i = 0; i < rows.size(); i++) {
    Record record;
    rc = make_record(static_cast<int>(rows[i].size()), rows[i].data(), record);
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to make record. ret=%d", rc);
      return rc;
    }
    rc = insert_record(record);
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to insert record. ret=%d", rc);
      return rc;
    }
  }
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to insert record. ret=%d", rc);
    // 插入数据失败，那么应该删除这个表
    rc = drop(db, table_id, path, name, base_dir);
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to drop table after insert record failed. ret=%d", rc);
    }
    return rc;
  }
  return rc;
}

RC Table::drop(Db *db, int32_t table_id, const char *path, const char *name, const char *base_dir)
{
  if (table_id < 0) {
    LOG_WARN("invalid table id. table_id=%d, table_name=%s", table_id, name);
    return RC::INVALID_ARGUMENT;
  }
  // 表名不能为空
  if (common::is_blank(name)) {
    LOG_WARN("Name cannot be empty.");
    return RC::INVALID_ARGUMENT;
  }

  // 实际上需要删除三个文件.index，.table以及.data，但是目前好像不需要删除文件，所以就先不用管
  // .index文件可能不止有一个所以需要借助drop_all_index函数

  db_       = db;
  base_dir_ = base_dir;
  RC rc1    = drop_all_index(name);
  if (rc1 != RC::SUCCESS) {
    LOG_ERROR("Failed to drop all indexes, table:%s", name);
    return rc1;
  }
  string table_meta_file_path = table_meta_file(base_dir, name);
  string table_data_file_path = table_data_file(base_dir, name);
  // delete file
  BufferPoolManager &bpm = db->buffer_pool_manager();
  RC                 rc  = bpm.delete_file(table_data_file_path);
  if (std::filesystem::remove(table_meta_file_path) && rc == RC::SUCCESS) {
    LOG_INFO("Successfully drop table %s:%s", base_dir, name);
  } else {
    LOG_WARN("Failed to drop table due to not exist!");
    return RC::SCHEMA_TABLE_NOT_EXIST;
  }

  // 似乎还需要处理一下buffer_pool
  return rc;
}

RC Table::open(Db *db, const char *meta_file, const char *base_dir)
{
  // 加载元数据文件
  fstream fs;
  string  meta_file_path = string(base_dir) + common::FILE_PATH_SPLIT_STR + meta_file;
  fs.open(meta_file_path, ios_base::in | ios_base::binary);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open meta file for read. file name=%s, errmsg=%s", meta_file_path.c_str(), strerror(errno));
    return RC::IOERR_OPEN;
  }
  if (table_meta_.deserialize(fs) < 0) {
    LOG_ERROR("Failed to deserialize table meta. file name=%s", meta_file_path.c_str());
    fs.close();
    return RC::INTERNAL;
  }
  fs.close();

  db_       = db;
  base_dir_ = base_dir;

  // 加载数据文件
  RC rc = init_record_handler(base_dir);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to open table %s due to init record handler failed.", base_dir);
    // don't need to remove the data_file
    return rc;
  }

  const int index_num = table_meta_.index_num();
  for (int i = 0; i < index_num; i++) {
    const IndexMeta          *index_meta        = table_meta_.index(i);
    vector<string>            index_meta_fields = index_meta->fields();
    vector<const FieldMeta *> fields_meta;
    for (unsigned i = 0; i < index_meta_fields.size(); i++) {
      const FieldMeta *field_meta = table_meta_.field(index_meta_fields[i].c_str());
      if (field_meta == nullptr) {
        LOG_ERROR("Found invalid index meta info which has a non-exists field. table=%s, index=%s, fields_size=%d",
                name(), index_meta->name(), index_meta->fields().size());
        // skip cleanup
        //  do all cleanup action in destructive Table function
        return RC::INTERNAL;
      }
      fields_meta.push_back(field_meta);
    }

    BplusTreeIndex *index      = new BplusTreeIndex();
    string          index_file = table_index_file(base_dir, name(), index_meta->name());

    rc = index->open(this, index_file.c_str(), *index_meta, fields_meta);
    if (rc != RC::SUCCESS) {
      delete index;
      LOG_ERROR("Failed to open index. table=%s, index=%s, file=%s, rc=%s",
                name(), index_meta->name(), index_file.c_str(), strrc(rc));
      // skip cleanup
      //  do all cleanup action in destructive Table function.
      return rc;
    }
    indexes_.push_back(index);
  }

  return rc;
}

RC Table::insert_record(Record &record)
{
  RC rc = RC::SUCCESS;

  // 这里加一个MAX_COLUMN_SIZE是为了记录每一列是否为NULL
  rc = record_handler_->insert_record(record.data(), table_meta_.record_size() + MAX_COLUMN_SIZE, &record.rid());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Insert record failed. table name=%s, rc=%s", table_meta_.name(), strrc(rc));
    return rc;
  }

  rc = insert_entry_of_indexes(record.data(), record.rid());
  if (rc != RC::SUCCESS) {  // 可能出现了键值重复
    // RC rc2 = delete_entry_of_indexes(record.data(), record.rid(), false /*error_on_not_exists*/);
    // if (rc2 != RC::SUCCESS) {
    //   LOG_ERROR("Failed to rollback index data when insert index entries failed. table name=%s, rc=%d:%s",
    //             name(), rc2, strrc(rc2));
    // }
    RC rc2 = record_handler_->delete_record(&record.rid());
    if (rc2 != RC::SUCCESS) {
      LOG_PANIC("Failed to rollback record data when insert index entries failed. table name=%s, rc=%d:%s",
                name(), rc2, strrc(rc2));
    }
  }
  return rc;
}

RC Table::visit_record(const RID &rid, function<bool(Record &)> visitor)
{
  return record_handler_->visit_record(rid, visitor);
}

RC Table::get_record(const RID &rid, Record &record)
{
  RC rc = record_handler_->get_record(rid, record);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to visit record. rid=%s, table=%s, rc=%s", rid.to_string().c_str(), name(), strrc(rc));
    return rc;
  }

  return rc;
}

RC Table::recover_insert_record(Record &record)
{
  RC rc = RC::SUCCESS;
  rc    = record_handler_->recover_insert_record(record.data(), table_meta_.record_size(), record.rid());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Insert record failed. table name=%s, rc=%s", table_meta_.name(), strrc(rc));
    return rc;
  }

  rc = insert_entry_of_indexes(record.data(), record.rid());
  if (rc != RC::SUCCESS) {  // 可能出现了键值重复
    RC rc2 = delete_entry_of_indexes(record.data(), record.rid(), false /*error_on_not_exists*/);
    if (rc2 != RC::SUCCESS) {
      LOG_ERROR("Failed to rollback index data when insert index entries failed. table name=%s, rc=%d:%s",
                name(), rc2, strrc(rc2));
    }
    rc2 = record_handler_->delete_record(&record.rid());
    if (rc2 != RC::SUCCESS) {
      LOG_PANIC("Failed to rollback record data when insert index entries failed. table name=%s, rc=%d:%s",
                name(), rc2, strrc(rc2));
    }
  }
  return rc;
}

const char *Table::name() const { return table_meta_.name(); }

const TableMeta &Table::table_meta() const { return table_meta_; }

RC Table::make_record(int value_num, const Value *values, Record &record)
{

  // 首先检查字段数量有没有超出上限
  if (value_num > MAX_COLUMN_SIZE * 8) {
    LOG_WARN("The number of fields exceeds the limit. table name:%s", table_meta_.name());
    return RC::INVALID_ARGUMENT;
  }

  RC rc = RC::SUCCESS;
  // 检查字段数量是否一致
  if (value_num + table_meta_.sys_field_num() != table_meta_.field_num()) {
    LOG_WARN("Input values don't match the table's schema, table name:%s", table_meta_.name());
    return RC::SCHEMA_FIELD_MISSING;
  }

  const int normal_field_start_index = table_meta_.sys_field_num();
  // 复制所有字段的值

  // 最后的MAX_COLUMN_SIZE是为了记录每一列是否为NULL
  int   record_size = table_meta_.record_size() + MAX_COLUMN_SIZE;
  char *record_data = (char *)malloc(record_size);
  // 这里的bitmap是用来记录每一列是否为NULL
  char *bitmap = record_data + table_meta_.record_size();
  memset(record_data, 0, record_size);

  for (int i = 0; i < value_num && OB_SUCC(rc); i++) {
    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
    const Value     &value = values[i];
    if (field->type() != value.attr_type()) {
      // 如果是NULL，其实可以适配所有类型，这里要做一个转换
      if (value.is_null() && field->nullable()) {
        switch (field->type()) {
          case AttrType::INTS: {
            Value val(0);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::BOOLEANS: {
            Value val(false);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::DATES: {
            Value val(20020627, 0);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::FLOATS: {
            Value val(0.0f);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::CHARS: {
            Value val("NULL", 4);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          default: {
            LOG_WARN("Unsupported type");
            return RC::INVALID_ARGUMENT;
          }
        }
        Value::set_bit(bitmap, i);
        continue;
      } else if (!field->nullable() && value.is_null()) {
        LOG_WARN("Field %s is not nullable", field->name());
        return RC::INVALID_ARGUMENT;
      }

      // 这里就是类型不符合，但是尝试类型转换
      Value real_value;
      rc = Value::cast_to(value, field->type(), real_value);
      if (OB_FAIL(rc)) {
        LOG_WARN("failed to cast value. table name:%s,field name:%s,value:%s ",
            table_meta_.name(), field->name(), value.to_string().c_str());
        break;
      }
      rc = set_value_to_record(record_data, real_value, field);
    } else {
      rc = set_value_to_record(record_data, value, field);
    }
  }
  if (OB_FAIL(rc)) {
    LOG_WARN("failed to make record. table name:%s", table_meta_.name());
    free(record_data);
    return rc;
  }

  record.set_data_owner(record_data, record_size);
  return RC::SUCCESS;
}

RC Table::make_new_record(
    int value_num, Value *values, Record &record, vector<FieldMeta> field_meta, vector<Value> value)
{
  RC rc = RC::SUCCESS;
  // 检查字段数目是否一致
  // NOTE: 这里的value_num是我们通过表扫描得到的，因此可能会在trx中含有系统字段，因此先减去系统字段的数量，后面就跟make
  // record函数的逻辑一样了
  value_num = value_num - table_meta_.sys_field_num();
  if (value_num + table_meta_.sys_field_num() != table_meta_.field_num()) {
    LOG_WARN("Input values don't match the table's schema, table name:%s", table_meta_.name());
    return RC::SCHEMA_FIELD_MISSING;
  }

  const int normal_field_start_index = table_meta_.sys_field_num();
  // 最后的MAX_COLUMN_SIZE是为了记录每一列是否为NULL
  int   record_size = table_meta_.record_size() + MAX_COLUMN_SIZE;
  char *record_data = (char *)malloc(record_size);
  // 这里的bitmap是用来记录每一列是否为NULL
  char *bitmap = record_data + table_meta_.record_size();
  memset(record_data, 0, record_size);

  for (int i = 0; i < value_num && OB_SUCC(rc); i++) {
    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
    LOG_TRACE("-----------------------------------------%s", field->name());
    bool     flag = false;
    unsigned j    = 0;
    for (j = 0; j < field_meta.size(); j++) {
      if (strcmp(field->name(), field_meta[j].name()) == 0) {
        flag = true;
        break;
      }
    }
    Value myvalue;
    if (flag == true) {
      myvalue = value[j];
    } else {
      myvalue = values[i + normal_field_start_index];
    }
    const Value &value = myvalue;
    if (field->type() != value.attr_type() || value.is_null()) {
      // 如果是NULL，其实可以适配所有类型，这里要做一个转换
      if (value.is_null() && field->nullable()) {
        switch (field->type()) {
          case AttrType::INTS: {
            Value val(0);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::BOOLEANS: {
            Value val(false);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::DATES: {
            Value val(20020627, 0);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::FLOATS: {
            Value val(0.0f);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          case AttrType::CHARS: {
            Value val("NULL", 4);
            val.set_null_flag(true);
            rc = set_value_to_record(record_data, val, field);
          }; break;
          default: {
            LOG_WARN("Unsupported type");
            return RC::INVALID_ARGUMENT;
          }
        }
        Value::set_bit(bitmap, i);
        continue;
      } else if (!field->nullable() && value.is_null()) {
        LOG_WARN("Field %s is not nullable", field->name());
        return RC::INVALID_ARGUMENT;
      }

      Value real_value;
      rc = Value::cast_to(value, field->type(), real_value);
      if (OB_FAIL(rc)) {
        LOG_WARN("failed to cast value. table name:%s,field name:%s,value:%s ",
            table_meta_.name(), field->name(), value.to_string().c_str());
        break;
      }
      rc = set_value_to_record(record_data, real_value, field);
    } else {
      rc = set_value_to_record(record_data, value, field);
    }
  }
  if (OB_FAIL(rc)) {
    LOG_WARN("failed to make new record. table name:%s", table_meta_.name());
    free(record_data);
    return rc;
  }

  record.set_data_owner(record_data, record_size);
  return RC::SUCCESS;
}

RC Table::set_value_to_record(char *record_data, const Value &value, const FieldMeta *field)
{
  size_t       copy_len = field->len();
  const size_t data_len = value.length();
  if (field->type() == AttrType::CHARS) {
    if (copy_len > data_len) {
      copy_len = data_len + 1;
    }
  }
  memcpy(record_data + field->offset(), value.data(), copy_len);
  return RC::SUCCESS;
}

RC Table::init_record_handler(const char *base_dir)
{
  string data_file = table_data_file(base_dir, table_meta_.name());  // 得到.data的路径

  BufferPoolManager &bpm = db_->buffer_pool_manager();
  RC                 rc  = bpm.open_file(db_->log_handler(), data_file.c_str(), data_buffer_pool_);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to open disk buffer pool for file:%s. rc=%d:%s", data_file.c_str(), rc, strrc(rc));
    return rc;
  }

  record_handler_ = new RecordFileHandler(table_meta_.storage_format());

  rc = record_handler_->init(*data_buffer_pool_, db_->log_handler(), &table_meta_);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to init record handler. rc=%s", strrc(rc));
    data_buffer_pool_->close_file();
    data_buffer_pool_ = nullptr;
    delete record_handler_;
    record_handler_ = nullptr;
    return rc;
  }

  return rc;
}

RC Table::get_record_scanner(RecordFileScanner &scanner, Trx *trx, ReadWriteMode mode)
{
  RC rc = scanner.open_scan(this, *data_buffer_pool_, trx, db_->log_handler(), mode, nullptr);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("failed to open scanner. rc=%s", strrc(rc));
  }
  return rc;
}

RC Table::get_chunk_scanner(ChunkFileScanner &scanner, Trx *trx, ReadWriteMode mode)
{
  RC rc = scanner.open_scan_chunk(this, *data_buffer_pool_, db_->log_handler(), mode);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("failed to open scanner. rc=%s", strrc(rc));
  }
  return rc;
}

RC Table::drop_all_index(const char *table_name)
{
  // 下面遍历整个文件目录，获取要删除的表所拥有的索引
  vector<string> all_indexes;
  string         str_table_name(table_name);
  for (const auto &entry : std::filesystem::directory_iterator(base_dir_)) {
    if (entry.is_regular_file()) {
      string filename = entry.path().filename().string();
      string reg;
      for (unsigned i = 0; i < filename.size() && filename[i] != '-'; i++) {
        reg += filename[i];
      }
      if (reg != str_table_name) {
        continue;
      }
      if (filename.rfind(name(), 0) == 0 && filename.substr(filename.size() - 6) == ".index") {
        all_indexes.push_back(
            filename.substr(str_table_name.size() + 1, filename.size() - str_table_name.size() - 1 - 6));
        std::cout << all_indexes[0] << endl;
        LOG_INFO("-------------------FILENAME:%s", all_indexes[0].c_str());
      }
    }
  }
  if (all_indexes.size() == 0) {
    LOG_INFO("No index need to be droped!");
  } else {
    RC rc;
    for (int i = 0; i < static_cast<int>(all_indexes.size()); i++) {
      rc = drop_index(all_indexes[i].c_str(), table_name);
      if (rc != RC::SUCCESS) {
        LOG_ERROR("Failed to drop all indexes!");
        return rc;
      }
    }
    LOG_INFO("Successfully drop all indexes!");
  }
  return RC::SUCCESS;
}

RC Table::drop_index(const char *index_name, const char *table_name)
{
  if (common::is_blank(index_name)) {
    LOG_INFO("Invalid input arguments, table name is %s, index_name is blank.", name());
    return RC::INVALID_ARGUMENT;
  }
  BplusTreeIndex *index      = new BplusTreeIndex();
  string          index_file = table_index_file(base_dir_.c_str(), table_name, index_name);
  RC              rc         = index->drop(this, index_file.c_str());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to drop bplus tree index. file name = %s, rc = %d:%s", index_file.c_str(), rc, strrc(rc));
    return rc;
  }
  LOG_ERROR("Drop index %s success!", index_file.c_str());
  return rc;
}

RC Table::create_index(Trx *trx, vector<const FieldMeta *> fields_meta, const char *index_name, bool unique)
{
  if (common::is_blank(index_name) || 0 == fields_meta.size()) {
    LOG_INFO("Invalid input arguments, table name is %s, index_name is blank or attribute_name is blank", name());
    return RC::INVALID_ARGUMENT;
  }

  // loop here
  // 这里改变一下存储的逻辑？
  // 其实需要把好几个字段合并到一个索引之中去
  IndexMeta new_index_meta;
  RC        rc = new_index_meta.init(index_name, fields_meta, unique);
  if (rc != RC::SUCCESS) {
    LOG_INFO("Failed to init IndexMeta in table:%s, index_name:%s, fields_size:%d, is_unique : %d",
             name(), index_name, fields_meta.size(), unique);
    return rc;
  }

  // 创建索引相关数据
  BplusTreeIndex *index      = new BplusTreeIndex();
  string          index_file = table_index_file(base_dir_.c_str(), name(), index_name);

  rc = index->create(this, index_file.c_str(), new_index_meta, fields_meta);
  if (rc != RC::SUCCESS) {
    delete index;
    LOG_ERROR("Failed to create bplus tree index. file name=%s, rc=%d:%s", index_file.c_str(), rc, strrc(rc));
    return rc;
  }

  // 遍历当前的所有数据，插入这个索引
  RecordFileScanner scanner;
  rc = get_record_scanner(scanner, trx, ReadWriteMode::READ_ONLY);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to create scanner while creating index. table=%s, index=%s, rc=%s", 
             name(), index_name, strrc(rc));
    return rc;
  }

  Record record;
  while (OB_SUCC(rc = scanner.next(record))) {
    rc = index->insert_entry(record.data(), &record.rid());
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to insert record into index while creating index. table=%s, index=%s, rc=%s",
               name(), index_name, strrc(rc));
      return rc;
    }
  }
  if (RC::RECORD_EOF == rc) {
    rc = RC::SUCCESS;
  } else {
    LOG_WARN("failed to insert record into index while creating index. table=%s, index=%s, rc=%s",
             name(), index_name, strrc(rc));
    return rc;
  }
  scanner.close_scan();
  LOG_INFO("inserted all records into new index. table=%s, index=%s", name(), index_name);

  indexes_.push_back(index);

  /// 接下来将这个索引放到表的元数据中
  TableMeta new_table_meta(table_meta_);
  rc = new_table_meta.add_index(new_index_meta);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to add index (%s) on table (%s). error=%d:%s", index_name, name(), rc, strrc(rc));
    return rc;
  }

  /// 内存中有一份元数据，磁盘文件也有一份元数据。修改磁盘文件时，先创建一个临时文件，写入完成后再rename为正式文件
  /// 这样可以防止文件内容不完整
  // 创建元数据临时文件
  string  tmp_file = table_meta_file(base_dir_.c_str(), name()) + ".tmp";
  fstream fs;
  fs.open(tmp_file, ios_base::out | ios_base::binary | ios_base::trunc);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open file for write. file name=%s, errmsg=%s", tmp_file.c_str(), strerror(errno));
    return RC::IOERR_OPEN;  // 创建索引中途出错，要做还原操作
  }
  if (new_table_meta.serialize(fs) < 0) {
    LOG_ERROR("Failed to dump new table meta to file: %s. sys err=%d:%s", tmp_file.c_str(), errno, strerror(errno));
    return RC::IOERR_WRITE;
  }
  fs.close();

  // 覆盖原始元数据文件
  string meta_file = table_meta_file(base_dir_.c_str(), name());

  int ret = rename(tmp_file.c_str(), meta_file.c_str());
  if (ret != 0) {
    LOG_ERROR("Failed to rename tmp meta file (%s) to normal meta file (%s) while creating index (%s) on table (%s). "
              "system error=%d:%s",
              tmp_file.c_str(), meta_file.c_str(), index_name, name(), errno, strerror(errno));
    return RC::IOERR_WRITE;
  }

  table_meta_.swap(new_table_meta);

  LOG_INFO("Successfully added a new index (%s) on the table (%s)", index_name, name());
  return rc;
}

RC Table::delete_record(const RID &rid)
{
  RC     rc = RC::SUCCESS;
  Record record;
  rc = get_record(rid, record);
  if (OB_FAIL(rc)) {
    return rc;
  }

  return delete_record(record);
}

RC Table::delete_record(const Record &record)
{
  RC rc = RC::SUCCESS;
  for (Index *index : indexes_) {
    rc = index->delete_entry(record.data(), &record.rid());
    ASSERT(RC::SUCCESS == rc, 
           "failed to delete entry from index. table name=%s, index name=%s, rid=%s, rc=%s",
           name(), index->index_meta().name(), record.rid().to_string().c_str(), strrc(rc));
  }
  rc = record_handler_->delete_record(&record.rid());
  return rc;
}

RC Table::insert_entry_of_indexes(const char *record, const RID &rid)
{
  RC          rc     = RC::SUCCESS;
  const char *bitmap = record + table_meta_.record_size();
  for (Index *index : indexes_) {
    // 只有一个字段 && 是NULL字段
    bool has_null = false;
    for (unsigned i = 0; i < index->fields_meta().size(); i++) {
      if (Value::get_bit(bitmap, index->fields_meta()[i].field_id())) {
        has_null = true;
        break;
      }
    }
    // 直接不插入
    if (has_null) {
      continue;
    }

    rc = index->insert_entry(record, &rid);
    if (rc != RC::SUCCESS) {
      break;
    }
  }
  return rc;
}

RC Table::delete_entry_of_indexes(const char *record, const RID &rid, bool error_on_not_exists)
{
  RC          rc     = RC::SUCCESS;
  const char *bitmap = record + table_meta_.record_size();
  for (Index *index : indexes_) {
    // 只有一个字段 && 是NULL字段
    bool has_null = false;
    for (unsigned i = 0; i < index->fields_meta().size(); i++) {
      if (Value::get_bit(bitmap, index->fields_meta()[i].field_id())) {
        has_null = true;
        break;
      }
    }
    // 如果有NULL字段，直接不插入
    if (has_null) {
      continue;
    }
    rc = index->delete_entry(record, &rid);
    if (rc != RC::SUCCESS) {
      if (rc != RC::RECORD_INVALID_KEY || !error_on_not_exists) {
        break;
      }
    }
  }
  return rc;
}

Index *Table::find_index(const char *index_name) const
{
  for (Index *index : indexes_) {
    if (0 == strcmp(index->index_meta().name(), index_name)) {
      return index;
    }
  }
  return nullptr;
}
Index *Table::find_index_by_field(vector<const char *> fields_name) const
{
  const TableMeta &table_meta = this->table_meta();
  const IndexMeta *index_meta = table_meta.find_index_by_field(fields_name);
  if (index_meta != nullptr) {
    return this->find_index(index_meta->name());
  }
  return nullptr;
}

RC Table::sync()
{
  RC rc = RC::SUCCESS;
  for (Index *index : indexes_) {
    rc = index->sync();
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to flush index's pages. table=%s, index=%s, rc=%d:%s",
          name(),
          index->index_meta().name(),
          rc,
          strrc(rc));
      return rc;
    }
  }

  rc = data_buffer_pool_->flush_all_pages();
  LOG_INFO("Sync table over. table=%s", name());
  return rc;
}
